# +----------------------------------------------------------------------
# | ChatWork智能聊天办公系统
# +----------------------------------------------------------------------
# | 软件声明: 本系统并非自由软件,未经授权任何形式的商业使用均属非法。
# | 版权保护: 任何企业和个人不允许对程序代码以任何形式任何目的复制/分发。
# | 授权要求: 如有商业使用需求,请务必先与版权所有者取得联系并获得正式授权。
# +----------------------------------------------------------------------
# | Author: ChatWork Team <2474369941@qq.com>
# +----------------------------------------------------------------------
import os
import hmac
import json
import time
import pytz
import urllib
import base64
import hashlib
import aiofiles
import requests
import threading
import websocket
import websockets
from typing import Literal
from datetime import datetime
from urllib.parse import urlencode, quote
from openai import AsyncOpenAI


class OpenaiAudio:
    @classmethod
    async def speech(
            cls,
            input_text: str,
            file_path: str,
            api_key: str,
            base_url: str = "https://api.openai.com/v1",
            timeout: int = 300,
            max_retries: int = 2,
            speed: int = 1,
            model: Literal["tts-1", "tts-1-hd"] = "tts-1",
            voice: Literal["alloy", "echo", "fable", "onyx", "nova", "shimmer"] = "alloy",
            response_format: Literal["mp3", "opus", "aac", "flac", "wav", "pcm"] = "mp3"
    ):
        """ 文本转语音 """
        # 文件
        if os.path.exists(file_path):
            os.remove(file_path)

        # 发起
        client = AsyncOpenAI(base_url=base_url, api_key=api_key, timeout=timeout, max_retries=max_retries)
        response = await client.audio.speech.create(
            model=model,
            voice=voice,
            speed=speed,
            input=input_text,
            response_format=response_format,
        )
        response.write_to_file(file_path)

    @classmethod
    async def transcriptions(
            cls,
            file_path: str,
            api_key: str,
            base_url: str = "https://api.openai.com/v1",
            timeout: int = 300,
            max_retries: int = 2,
            model: Literal["whisper-1"] = "whisper-1"
    ):
        """ 语音转文本 """
        client = AsyncOpenAI(base_url=base_url, api_key=api_key, timeout=timeout, max_retries=max_retries)
        audio_file = open(file_path, "rb")
        transcription = await client.audio.transcriptions.create(
            model=model,
            file=audio_file
        )
        return transcription.text


class XunfeiAudio:
    class WssSpeech:
        """ 超拟人合成 """
        def __init__(self, APPID: str, APIKey: str, APISecret: str, Text: str, FilePath: str, **kwargs):
            self.APPID = APPID
            self.APIKey = APIKey
            self.APISecret = APISecret
            self.InputText = Text
            self.FilePath = FilePath
            self.uri = "wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6"

            self.CommonArgs = {"app_id": self.APPID, "status": 2}
            self.BusinessArgs = {
                "tts": {
                    "vcn": kwargs.get("vcn", "x4_lingxiaoxuan_oral"),  # 发音人参数
                    "volume": kwargs.get("volume", 50),  # 设置音量大小
                    "rhy": kwargs.get("rhy", 0),         # 返回拼音标注 [0=不返回拼音,1=返回拼音]
                    "speed": kwargs.get("speed", 50),    # 设置合成语速 [值越大,语速越快]
                    "pitch": kwargs.get("pitch", 50),    # 设置振幅高低 [该参数调整效果]
                    "bgs": kwargs.get("bgs", 0),         # 设置背景音乐 [0=无背景音,1=内置背景音1,2=内置背景音2]
                    "reg": kwargs.get("reg", 0),         # 英文发音方式 [0=自动,1=英文字母发音,2=字母朗读]
                    "rdn": kwargs.get("rdn", 0),         # 数字发音方式 [0=自动,1=数值,2=字符串,3=字符串优先]
                    "audio": {
                        "encoding": kwargs.get("encoding", "lame"),       # 合成音频的格式 [lame合成音频格式为mp3]
                        "sample_rate": kwargs.get("sample_rate", 24000),  # 合成音频采样率 [16000, 8000, 24000]
                        "bit_depth": kwargs.get("bit_depth", 16),         # 合成音频位深: [16, 8]
                        "channels": kwargs.get("channels", 1),            # 音频声道数值
                        "frame_size": kwargs.get("frame_size", 0),
                    }
                }
            }
            self.Data = {
                "text": {
                    "encoding": "utf8",
                    "compress": "raw",
                    "format": "plain",
                    "status": 2,
                    "seq": 0,
                    "text": str(base64.b64encode(self.InputText.encode("utf-8")), "UTF8")
                }
            }

        @classmethod
        def _sha256base64(cls, data):
            sha256 = hashlib.sha256()
            sha256.update(data)
            digest = base64.b64encode(sha256.digest()).decode(encoding='utf-8')
            return digest

        @classmethod
        def _parse_url(cls, request_url):
            stIdx = request_url.index("://")
            host = request_url[stIdx + 3:]
            schema = request_url[:stIdx + 3]
            edIdx = host.index("/")
            if edIdx <= 0:
                raise Exception("invalid request url:" + request_url)
            path = host[edIdx:]
            host = host[:edIdx]
            return {"host": host, "path": path, "schema": schema}

        def _auth_url(self, request_url: str):
            api_key: str = self.APIKey
            api_secret: str = self.APISecret

            u = self._parse_url(request_url)
            host = u["host"]
            path = u["path"]
            now = datetime.now(pytz.utc)
            date = now.strftime("%a, %d %b %Y %H:%M:%S GMT")

            signature_origin: str = "host: {}\ndate: {}\n{} {} HTTP/1.1".format(host, date, "GET", path)
            hmac_key: bytes = api_secret.encode('utf-8')
            hmac_msg: bytes = signature_origin.encode('utf-8')
            signature_sha = hmac.new(hmac_key, hmac_msg, digestmod=hashlib.sha256).digest()
            signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
            author_params = (api_key, "hmac-sha256", "host date request-line", signature_sha)
            author_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % author_params
            authorization = base64.b64encode(author_origin.encode('utf-8')).decode(encoding='utf-8')
            values = {
                "host": host,
                "date": date,
                "authorization": authorization
            }
            return request_url + "?" + urlencode(values)

        async def websocket_client(self):
            url: str = self._auth_url(self.uri)
            try:
                async with websockets.connect(url) as ws:
                    await self.on_open(ws)
                    async for message in ws:
                        await self.on_message(ws, message)
            except websockets.exceptions.ConnectionClosed as e:
                raise Exception(400, f"Connection closed: {e}")
            except websockets.exceptions.InvalidHeader as e:
                raise Exception(400, f"Invalid header: {e}")
            except websockets.exceptions.InvalidStatus as e:
                errMsg = json.loads(e.response.body.decode())
                raise Exception(e.response.status_code, errMsg["message"])
            except Exception as e:
                if isinstance(e.args, tuple):
                    error = self.format_error(int(e.args[0]), str(e))
                    raise Exception(500, error)
                raise Exception(500, f"An unexpected error occurred: {e}")

        async def on_open(self, ws):
            await ws.send(json.dumps({
                "header": self.CommonArgs,
                "parameter": self.BusinessArgs,
                "payload": self.Data
            }))

            if os.path.exists(self.FilePath):
                os.remove(self.FilePath)

        async def on_message(self, ws, message):
            message = json.loads(message)
            code = message["header"]["code"]
            if "payload" in message:
                audio = message["payload"]["audio"]['audio']
                audio = base64.b64decode(audio)
                status = message["payload"]['audio']["status"]
                if status == 2:
                    await ws.close()
                if code != 0:
                    await ws.close()
                    errMsg = message["message"]
                    raise Exception(errMsg["code"], errMsg["message"])
                else:
                    async with aiofiles.open(self.FilePath, "ab") as f:
                        await f.write(audio)
            elif "header" in message:
                header = message["header"]
                if header["code"] != 0:
                    await ws.close()
                    raise Exception(header["code"], header["message"])

        @classmethod
        def format_error(cls, code: int, error: str):
            error = {
                10009: "10009: 输入数据非法",
                10010: "10010: 没有授权许可或授权数已满: 请到讯飞提交工单",
                10019: "10019: session超时: 检查是否数据发送完毕但未关闭连接",
                10043: "10043: 音频解码失败",
                10139: "10139: 参数错误: 检查参数是否正确",
                10222: "10222: 上传的数据超过了接口上限 或 SSL证书无效",
                10223: "10223: lb 找不到节点: 请到讯飞提交工单",
                10313: "10313: appid和apikey不匹配: 检查appid是否合法",
                11200: "11200: 功能未授权: appid错误/功能未授权/超出调用总量"
            }
            return error.get(code, error)

    class WssTransfer:
        def __init__(self, APPID: str, ApiKey: str):
            self.appid = APPID
            self.apiKey = ApiKey
            self.base_url = "ws://rtasr.xfyun.cn/v1/ws"

            ts = str(int(time.time()))
            tt = (self.appid + ts).encode("utf-8")
            md5 = hashlib.md5()
            md5.update(tt)
            baseString = md5.hexdigest()
            baseString = bytes(baseString, encoding="utf-8")

            apiKey = self.apiKey.encode("utf-8")
            signa = hmac.new(apiKey, baseString, hashlib.sha1).digest()
            signa = base64.b64encode(signa)
            signa = str(signa, "utf-8")

            self.end_tag = "{\"end\": true}"
            self.wss_url = self.base_url + "?appid=" + self.appid + "&ts=" + ts + "&signa=" + quote(signa)
            self.ws = websocket.create_connection(self.wss_url)
            print("坎坎坷坷")
            print(self.ws)
            self.tRecv = threading.Thread(target=self.recv)
            self.tRecv.start()

        def send(self, file_path):
            print("开始吧少年")
            file_object = open(file_path, "rb")
            try:
                index = 1
                while True:
                    chunk = file_object.read(1280)
                    if not chunk:
                        break
                    self.ws.send(chunk)
                    index += 1
                    time.sleep(0.04)
            finally:
                file_object.close()
            print("借古讽今赶紧给人家")
            self.ws.send(bytes(self.end_tag.encode("utf-8")))
            print("send end tag success")

        def recv(self):
            try:
                while self.ws.connected:
                    result = str(self.ws.recv())
                    if len(result) == 0:
                        print("receive result end")
                        break
                    result_dict = json.loads(result)
                    # 解析结果
                    if result_dict["action"] == "started":
                        print("handshake success, result: " + result)

                    if result_dict["action"] == "result":
                        result_1 = result_dict
                        # result_2 = json.loads(result_1["cn"])
                        # result_3 = json.loads(result_2["st"])
                        # result_4 = json.loads(result_3["rt"])
                        print("rtasr result: " + result_1["data"])

                    if result_dict["action"] == "error":
                        print("rtasr error: " + result)
                        self.ws.close()
                        return
            except websocket.WebSocketConnectionClosedException:
                print("receive result end")

        def close(self):
            print("结束了")
            self.ws.close()
            print("connection closed")

    class AsyncTransfer:
        def __init__(self, APPID: str, SecretKey: str):
            self.appid = APPID
            self.secretKey = SecretKey
            self.ts = str(int(time.time()))
            self.signa = self.get_signa()

            self.lFasr_host = "https://raasr.xfyun.cn/v2/api"
            self.api_upload = "/upload"
            self.api_get_result = "/getResult"

        def get_signa(self):
            m2 = hashlib.md5()
            m2.update((self.appid + self.ts).encode('utf-8'))
            md5 = m2.hexdigest()
            md5 = bytes(md5, encoding='utf-8')
            signa = hmac.new(self.secretKey.encode('utf-8'), md5, hashlib.sha1).digest()
            signa = base64.b64encode(signa)
            signa = str(signa, 'utf-8')
            return signa

        def upload(self, filePath: str):
            file_len = os.path.getsize(filePath)
            file_name = os.path.basename(filePath)
            param_dict = urllib.parse.urlencode({
                "appId": self.appid,
                "signa": self.signa,
                "ts": self.ts,
                "fileSize": file_len,
                "fileName": file_name,
                "duration": "200"
            })

            data = open(filePath, "rb").read(file_len)
            response = requests.post(
                url=self.lFasr_host + self.api_upload + "?" + param_dict,
                headers={"Content-type": "application/json"},
                data=data
            )

            result = json.loads(response.text)
            return result

        def get_result(self, orderId):
            param_dict = urllib.parse.urlencode({
                "appId": self.appid,
                "signa": self.signa,
                "ts": self.ts,
                "orderId": orderId,
                "resultType": "transfer"
            })

            result = None
            status = 3
            while status == 3:
                response = requests.post(
                    url=self.lFasr_host + self.api_get_result + "?" + param_dict,
                    headers={"Content-type": "application/json"}
                )

                result = json.loads(response.text)
                status = result["content"]["orderInfo"]["status"]
                if status == 4:
                    break
                time.sleep(20)

            return result

    @classmethod
    async def speech(cls, input_text: str, file_path: str, app_id: str, api_key: str, api_secret: str, **kwargs):
        """
        文本转语音 (超拟人合成V3)
        文档的地址: https://www.xfyun.cn/doc/spark/super%20smart-tts.html

        Args:
            input_text (str): 需要转换语音的文本
            file_path (str): 保存音频的路径(如: /www/demo.mp3)
            app_id (str): appId
            api_key (str): apiKey
            api_secret (str): apiSecret

        Author:
            zero
        """
        client = cls.WssSpeech(
            APPID=app_id,
            APIKey=api_key,
            APISecret=api_secret,
            Text=input_text,
            FilePath=file_path,
            kwargs=kwargs
        )

        await client.websocket_client()

    @classmethod
    async def transcriptions(cls, file_path: str, app_id: str, secret_key: str, mode: str = "async"):
        """
        语音转文本 (语言转写/实时语音转写)
        文档的地址: https://www.xfyun.cn/doc/asr/ifasr_new/API.html

        Args:
            file_path (str): 语音文件路径
            app_id (str): APPID
            secret_key (str): SecretKey
            mode (str): SecretKey

        Author:
            zero
        """
        if mode == "async":
            client = cls.AsyncTransfer(APPID=app_id, SecretKey=secret_key)
            upload = client.upload(file_path)
            result = client.get_result(upload["content"]["orderId"])
        elif mode == "wss":
            client = cls.WssTransfer(APPID=app_id, ApiKey=secret_key)
            client.send(file_path)


class QwenAudio:

    @classmethod
    async def speech(cls):
        import dashscope
        from dashscope.audio.tts_v2 import SpeechSynthesizer

        """
        https://help.aliyun.com/zh/model-studio/user-guide/text-to-speech
        """

        dashscope.api_key = 'YOUR-DASHSCOPE-API-KEY'
        model = "cosyvoice-v1"
        voice = "longxiaochun"

        synthesizer = SpeechSynthesizer(model=model, voice=voice)
        audio = synthesizer.call("今天天气怎么样？")
        print('requestId: ', synthesizer.get_last_request_id())
        with open('output.mp3', 'wb') as f:
            f.write(audio)

    @classmethod
    async def transcriptions(cls):
        import dashscope
        from urllib import request
        from http import HTTPStatus

        dashscope.api_key = 'YOUR-DASHSCOPE-API-KEY'

        task_response = dashscope.audio.asr.Transcription.async_call(
            model='paraformer-v2',
            file_urls=[
                'https://dashscope.oss-cn-beijing.aliyuncs.com/samples/audio/paraformer/hello_world_female2.wav',
                'https://dashscope.oss-cn-beijing.aliyuncs.com/samples/audio/paraformer/hello_world_male2.wav'
            ],
            language_hints=['zh', 'en']
        )

        transcription_response = dashscope.audio.asr.Transcription.wait(task=task_response.output.task_id)
        if transcription_response.status_code == HTTPStatus.OK:
            for transcription in transcription_response.output['results']:
                url = transcription['transcription_url']
                result = json.loads(request.urlopen(url).read().decode('utf8'))
                print(json.dumps(result, indent=4, ensure_ascii=False))
            print('transcription done!')
        else:
            print('Error: ', transcription_response.output.message)
